package juc;

import java.util.Date;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.IntStream;

import static java.lang.Thread.currentThread;
import static java.util.concurrent.ThreadLocalRandom.current;

/**
 * Phaser JDK 1.7+
 * 可被重复使用的同步屏障；
 * 类似CyclicBarrier和CountDownLatch的合集；
 * 更灵活、使用挑战更大！
 * phaser n.  移相器;
 * @author ben
 * @date 2021-09-20 20:14:24 CST
 */
public class TestPhaser {

	public static Consumer<Object> cs = item->{
		Date now = new Date();
		if (null == item || item == "") {
			System.out.println("");
		} else {
			System.out.println("[" + now + "]" + item);
		}
	};

	public static void main(String[] args) throws InterruptedException {
//		test1();
		
//		test2();
		
//		test3();
		
//		test4();
		
//		test5();
		
		test6();
	}
	
	/**
	 * 试验1：当做CountDownLatch使用
	 * 子线程调用 非阻塞的arrive()
	 * CountDownLatch 一次性的，而 Phaser可以重复使用
	 * 主线程等待5个子线程完成任务
	 * @author ben
	 * @date 2021-09-20 20:14:44 CST
	 */
	public static void test1() {
		String tnm = currentThread().getName();
		
		cs.accept(tnm + ": test1() start...");
		
		Phaser ph = new Phaser();
		cs.accept(tnm + ": ph=" + ph);
		
		// 启用多个线程执行任务
		IntStream.range(0, 5).forEach(it->{
			new Thread(()->{
				String tn = currentThread().getName();
				// 注册
				ph.register();
				cs.accept(tn + ": start, ph=" + ph);
				
				// 休眠不同时间
				final int bound = 15;
				int slp = current().nextInt(bound);
				cs.accept(tn + ": slp=" + slp);
				try {
					TimeUnit.SECONDS.sleep(slp);
				} catch (InterruptedException e) {
					cs.accept(tn + ": InterruptedException 1");
				}
				
				// 任务完成：非阻塞
				ph.arrive();
				
				cs.accept(tn + ": end, ph=" + ph);
			}, "T" + it).start();
		});
		
		// 休眠：确保子线程都启动后主线程再执行
		try {
			TimeUnit.SECONDS.sleep(1);
		} catch (InterruptedException e) {
			cs.accept(tnm + ": InterruptedException");
		}
		
		// 第一次注册
		int i1 = ph.register();
		cs.accept(tnm + ": i1=" + i1 + ", ph=" + ph);
		
//		// 异常操作1：第二次注册
//		i1 = ph.register();
//		cs.accept(tnm + ": i1=" + i1 + ", ph=" + ph);
//		
//		// 解决异常操作1 的方式
//		// 调用arrive()，则不会阻塞，程序执行完成
//		int iar = ph.arrive();
//		cs.accept(tnm + ": iar=" + iar + ", ph=" + ph);
		
		// 前面注册2次的话，这里会阻塞
		int i2 = ph.arriveAndAwaitAdvance();
		// 完成后，ph的phase值加1 变为 1
		cs.accept(tnm + ": i2=" + i2 + ", ph=" + ph);
		cs.accept(tnm + ": test1() end.");
	}
	
	/**
	 * 试验2：模拟CyclicBarrier使用
	 * 子线程也调用 阻塞的 arriveAndAwaitAdvance()：到达屏障，一起结束
	 * @author ben
	 * @throws InterruptedException 
	 * @date 2021-09-21 20:40:54 CST
	 */
	public static void test2() throws InterruptedException {
		String tnm = currentThread().getName();
		cs.accept(tnm + ": test2() start...");
		
		Phaser ph = new Phaser();
		cs.accept(tnm + ": ph=" + ph);
		
		// 多个子线程
		IntStream.range(0, 5).forEach(item->{
			new Thread(()->{
				String tn = currentThread().getName();
				ph.register();
				cs.accept(tn + ": start ph=" + ph);
				try {
					int slp = current().nextInt(15);
					cs.accept(tn + ": slp=" + slp);
					TimeUnit.SECONDS.sleep(slp);
					cs.accept(tn + ": after sleep");
					ph.arriveAndAwaitAdvance();
					cs.accept(tn + ": ph=" + ph);
				} catch (InterruptedException e) {
					cs.accept(tn + ": InterruptedException");
				}
				cs.accept(tn + ": end");
			}, "T" + item).start();
		});
		
		TimeUnit.SECONDS.sleep(1);
		
		ph.register();
		
		ph.arriveAndAwaitAdvance();
		
		cs.accept(tnm + ": test2() end.");
	}

	/**
	 * 试验3：重写onAdvance方法
	 * 模拟 CyclicBarrier 的 CyclicBarrier(int parties, Runnable barrierAction) 构造函数
	 * @author ben
	 * @date 2021-09-21 20:59:27 CST
	 */
	public static void test3() {
		String tnm = currentThread().getName();
		cs.accept(tnm + ": test3() start...");
		
		final Test3Try ttry = new Test3Try(()->{
			cs.accept("所有子任务已完成工作");
		});
		
		// 子线程处理任务
		IntStream.range(0, 5).forEach(item->{
			new Thread(()->{
				String tn = currentThread().getName();
				cs.accept(tn + ": start");
				
				ttry.register();
				cs.accept(tn + ": ttry=" + ttry);
				try {
					int slp = current().nextInt(15);
					cs.accept(tn + ": slp=" + slp);
					TimeUnit.SECONDS.sleep(slp);
					cs.accept(tn + ": after sleep");
					ttry.arriveAndAwaitAdvance();
					cs.accept(tn + ": ttry=" + ttry);
				} catch (InterruptedException e) {
					cs.accept(tn + ": InterruptedException");
				}
				cs.accept(tn + ": end");
			}, "T" + item).start();
		});

		// 没有 注册、等待 phaser
		
		cs.accept(tnm + ": test3() end");
	}
	
	/**
	 * 试验4：重写onAdvance方法-2
	 * @author ben
	 * @date 2021-09-22 19:19:45 CST
	 */
	public static void test4() {
		String tnm = currentThread().getName();
		cs.accept(tnm + ": test4() start...");
		
		Phaser ph = new Phaser(2) {
			
			@Override
			protected boolean onAdvance(int phase, int registeredParties) {
				System.out.println("onAdvance this=" + this);
				return phase > 1;
			}
		};
		
		cs.accept("1、" + ph.arrive());
		cs.accept("ph=" + ph);
		cs.accept("2、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("");
		
		cs.accept("3、" + ph.arrive());
		cs.accept("ph=" + ph);
		cs.accept("4、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("");
		
		cs.accept("5、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("6、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("");
		
		// phaser已经终结了，下面的调用没用，且返回值为负数
		int rt = ph.arriveAndAwaitAdvance();
		cs.accept("rt=" + rt + ", pre phase=" + (rt + Integer.MAX_VALUE));
		
		cs.accept("7、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("8、" + ph.arrive());
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		cs.accept("");
		
		cs.accept(tnm + ": test4() end.");
	}

	/**
	 * 试验5：因onAdvance被阻塞
	 * 1、最后一个arrive的会被阻塞——等onAdvance执行完毕
	 * 2、执行onAdvance期间执行register时被阻塞
	 * @author ben
	 * @date 2021-09-22 19:37:45 CST
	 */
	public static void test5() {
		String tnm = currentThread().getName();
		cs.accept(tnm + ": test5() start...");

		Phaser ph = new Phaser(2) {
			
			@Override
			protected boolean onAdvance(int phase, int registeredParties) {
				cs.accept("onAdvance this=" + this);
				
				try {
					TimeUnit.SECONDS.sleep(30L);
					cs.accept("onAdvance after sleep");
				} catch (InterruptedException e) {
					cs.accept("onAdvance: InterruptedException");
				}
				
				return phase > 1;
			}
		};
		
		cs.accept("1、" + ph.arrive());
		cs.accept("ph=" + ph);
		
		// 第二次调用arrive(),被阻塞，等待onAdvance执行完毕
//		cs.accept("2、" + ph.arrive());
//		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
//		cs.accept("");
		
		// 新线程执行第二次arrive()，会被阻塞
		new Thread(()->{
			String tname = currentThread().getName();
			cs.accept(tname + ": arrive() 1");
			ph.arrive();
			cs.accept(tname + ": arrive() 2");
		}).start();
		
		// 确保上面线程先启动
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			cs.accept("InterruptedException e");
		}
		
		// 在前一个phase的onAdvance执行期间，执行register()，阻塞
		cs.accept("before register() ph=" + ph);
		ph.register();
		cs.accept("after register() ph=" + ph);
		
		cs.accept(tnm + ": test5() end.");
	}
	
	/**
	 * 试验6：其它函数
	 * @author ben
	 * @date 2021-09-22 19:55:21 CST
	 */
	public static void test6() {
		String tnm = currentThread().getName();
		cs.accept(tnm + ": test5() start...");

		Phaser ph = new Phaser(2);
		cs.accept("ph=" + ph);
		
		cs.accept("1、rt=" + ph.arriveAndDeregister());
		cs.accept("ph=" + ph);
		cs.accept("2、rt=" + ph.arriveAndDeregister());
		// parties==0，此时，phaser 被终结！
		cs.accept("ph=" + ph + ", isTerminated=" + ph.isTerminated());
		
		// 无法修复被终结的phaser！
		int rt = ph.bulkRegister(2);
		cs.accept("rt=" + rt);
		cs.accept("ph=" + ph);
		cs.accept("");
		
		Phaser ph2 = new Phaser();
		cs.accept("ph2=" + ph2 + ", isTerminated=" + ph2.isTerminated());
		
		// parties==0 此时执行 arrive()等发生异常
//		cs.accept("ph2 1.rt=" + ph2.arrive());
//		cs.accept("ph2 1.rt=" + ph2.arriveAndAwaitAdvance());
//		cs.accept("ph2=" + ph2 + ", isTerminated=" + ph2.isTerminated());
		
		ph2.bulkRegister(2);
		cs.accept("ph2=" + ph2 + ", isTerminated=" + ph2.isTerminated());
		for (int i=0; i<10; i++) {
			ph2.arrive();
			ph2.arrive();
			cs.accept("ph2=" + ph2 + ", isTerminated=" + ph2.isTerminated());
		}

		cs.accept(tnm + ": test5() end.");
	}
	
}

/**
 * 试验3使用：继承Phaser类，重写其onAdvance方法
 * @author ben
 * @date 2021-09-21 21:08:16 CST
 */
class Test3Try extends Phaser {
	
	private final Runnable run;
	
	public Test3Try(Runnable run) {
		super();
		this.run = run;
	}
	
	// 重写onAdvance：当parties个任务到达某个phase时，该方法将被执行
	@Override
	protected boolean onAdvance(int phase, int registeredParties) {
		this.run.run();
		return super.onAdvance(phase, registeredParties);
	}
}
